package com.rose.doorto.footbathdoor.config;

import com.rose.doorto.footbathdoor.enums.MessageStatusEnum;
import com.rose.doorto.footbathdoor.service.QueueMessagesService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.Map;

/**
 * @Author 李庆超
 * @email 1481791489@qq.com
 * @Package：com.rose.doorto.footbathdoor.config
 * @Project：door-to-door
 * @name：RabbitConfig
 * @date 2024/09/03 2024/9/3:14:31:
 * 开启rabbitmq的自动配置,并添加自定义配置
 * 如果只需要创建Exchange,queue,binging,发送消息等,只需要使用AmqpAdmin就足够。
 * 但如果要使用@RabbitListener监听消息(消费)，必须有@EnableRabbit开启功能,这个注解其实是往容器中注册了RabbitListenerAnnotationBeanPostProcessor
 *
 * 保证消息不丢失:
 * 1.开启publisher确认机制(confirmCallback,returnCallback)和consumer确认(手动ack/nack)
 * 2.每条发送的消息在数据库做好记录,定期扫描数据库将发送失败的消息重新发送。
 */
@EnableRabbit
@Slf4j
@Configuration
public class RabbitConfig {


    @Autowired
    private RabbitTemplate rabbitTemplate;


    @Autowired
    private QueueMessagesService queueMessagesService;

    /**
     * 如果发送的消息是一个对象，会使用序列化机制，由MessageConverter转换器处理，
     * 默认是WhiteListDeserializingMessageConverter，使用jdk序列化，所以这些bean必须实现Serializable接口
     *
     * 为了使用json序列化，我们需要往容器中添加一个使用json格式的MessageConverter，发送的消息会标记这个对象的全类名
     *
     * 监听队列的方法参数，可以使用Object obj来接收消息内容，通过obj.getClass()能看到真正类型是 org.springframework.amqp.core.Message
     * 所以也可以直接使用 Message me来接收，通过me.getBody()能够得到消息体
     * 如果知道消息体的本质类型，也可以直接使用 XXXEntity 来接收，
     * @return
     */
    @Bean
    MessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }


    /**
     * 定制RabbitTemplate
     * 1、服务收到消息回调
     *    1.开启配置文件 spring.rabbitmq.publisher-confirms=true
     *    2.设置确认回调
     */
    @PostConstruct
    public void setCallback(){

        /**
         * CorrelationData correlationData:  当前消息的唯一关联数据(认为是这个消息的唯一id)
         * boolean ack: 消息是否成功收到  只要消息成功抵达Broker, ack = true
         * @Nullable String cause:  失败的原因
         */
        RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean b, String cause) {
                if (b){
                    log.info("消息投递到交换机成功：[correlationData={}]",correlationData);
                    int i = queueMessagesService.updateStatusByMessageId(correlationData.getId(), MessageStatusEnum.SUCCESSSTATUS.getCode());
                    if (i>=1){
                        System.out.println("投递成功");
                    }else {
                        System.out.println("投递失败");
                    }
//                    queueMessagesMapper.update(new QueueMessages(),new UpdateWrapper<QueueMessages>().set("status", MessageStatusEnum.SUCCESSSTATUS.getCode()).eq("messageId",correlationData));
                }else {
                    int i = queueMessagesService.updateStatusByMessageId(correlationData.getId(), MessageStatusEnum.ERRORSTATUS.getCode());
                    log.error("消息投递到交换机失败：[correlationData={}，原因：{}]", correlationData, cause);
                }
            }
        };
        rabbitTemplate.setConfirmCallback(confirmCallback);
//        rabbitTemplate.setConfirmCallback((CorrelationData correlationData, boolean ack, String cause) -> {
//            if (ack) {
//                log.info("消息投递到交换机成功：[correlationData={}]",correlationData);
//            } else {
//                log.error("消息投递到交换机失败：[correlationData={}，原因：{}]", correlationData, cause);
//            }
//        });
        /**
         * 消息由Exchange路由到Queue失败回调    (比如路由键不对)
         * 只要消息没有投递给指定的队列,就触发这个失败回调
         * Message message: 投递失败的消息详细信息
         * int replyCode:   回复的状态码
         * String replyText: 回复的文本内容
         * String exchange:  当时这个消息发送给哪个交换机的
         * String routingKey： 当时这个消息用哪个路由键
         */
        rabbitTemplate.setReturnCallback((Message message, int replyCode, String replyText, String exchange, String routingKey) -> {
            log.error("路由到队列失败，消息内容：{}，交换机：{}，路由件：{}，回复码：{}，回复文本：{}", message, exchange, routingKey, replyCode, replyText);
        });
//        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
//            @Override
//            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
//                log.error("路由到队列失败，消息内容：{}，交换机：{}，路由件：{}，回复码：{}，回复文本：{}", message, exchange, routingKey, replyCode, replyText);
//            }
//        });
    }


    /**
     * 死信交换机
     * @return
     */
    @Bean
    public Exchange deadLetterExchangeName(){
        /**
         * String name:  交换机的名字
         * boolean durable:   是否持久化
         * boolean autoDelete:  是否自动删除
         * Map<String, Object> arguments:
         */
        return new TopicExchange("deadLetterExchangeName", true, false, null);
    }

    /**
     * 死信队列/延时队列
     * @return
     */
    @Bean
    public Queue DieQueue(){
        /**
         * String name:    队列的名字
         * boolean durable:  是否持久化
         * boolean exclusive:
         * boolean autoDelete,  是否自动删除
         * @Nullable Map<String, Object> arguments:
         */
        Map<String, Object> arguments = new HashMap<>();
        arguments.put("x-dead-letter-exchange", "deadLetterExchangeName");   //绑定的交换机
        arguments.put("x-dead-letter-routing-key", "routing_key");    //绑定所需要的路由键
        arguments.put("x-message-ttl", 3 * 60 * 1000);   //超时时间
//        channel.queueDeclare("deadLetterDoorQueue", true, false, false, arguments);
        Queue queue = new Queue("deadLetterDoorQueue", true, false, arguments.isEmpty());
        return queue;
    }

    /**
     * 第一个绑定关系,交换机路由到延时队列
     * @return
     */
    @Bean
    public Binding DieBinding(){
        /**
         * String destination:   目的地
         * DestinationType destinationType:  目的地类型
         * String exchange: 绑定的交换机名字
         * String routingKey:  所使用的路由键
         * @Nullable Map<String, Object> arguments:  其他绑定参数，例如消息的最大大小、消息的过期时间等。
         */
        return new Binding("deadLetterDoorQueue", Binding.DestinationType.QUEUE, "deadLetterExchangeName", "dieLetter", null);
    }

    /**
     * 第二个绑定关系,死掉的信息通过指定的路由键路由到另一个交换机
     * @return
     */
    @Bean
    public Binding DieExchangeBing(){
        /**
         * String destination:
         * DestinationType destinationType:
         * String exchange:
         * String routingKey:
         * @Nullable Map<String, Object> arguments:
         */
        return new Binding("deadLetterExchangeName", Binding.DestinationType.EXCHANGE, "deadLetterExchangeName", "dieLetter", null);
    }

    /**
     * 消息队列的使用场景:
     *    1.异步处理
     *    2.应用解耦
     *    3.流量控制
     * 1、大多数应用中,可通过消息服务中间件来提升系统异步通信、扩展解耦能力
     * 2、消息服务中两个重要概念:
     * 消息代理(message broker) 和目的地(destination)
     * 当消息发送者发送消息以后,将由消息代理接管,消息代理保证消息传递到指定目的地。
     * 3、消息队列主要有两种形式的目的地
     *    3.1) 队列(queue):  点对点消息通信(point-to-point)
     *    3.2) 主题(topic):  发布(publish)/订阅(subscribe)消息通信
     *    3.4) 点对点式: 消息发送者发送消息,消息代理将其放入一个队列中,消息接收者从队列中获取消息内容,消息读取后被移除队列。
     *                  消息只有唯一的发送者和接受者,但并不是说只能有一个接收者。
     *    3.5) 发布订阅式: 发送者（发布者）发送消息到主题，多个接收者（订阅者）监听（订阅）这个 主题，那么就会在消息到达时同时收到消息
     *    3.6) JMS（Java Message Service）JAVA消息服务：
     *         基于JVM消息代理的规范。ActiveMQ、HornetMQ是JMS实现
     *    3.7) AMQP（Advanced Message Queuing Protocol）
     *         高级消息队列协议，也是一个消息代理的规范，兼容JMS
     *         RabbitMQ是AMQP的实现
     *
     * RabbitMQ的确认机制----可靠抵达(保证消息不丢失)
     * 消息可能会在三个阶段丢失:  p->b、e->q、q->c
     * publisher confirmCallback 确认机制  p端成功投递到b端触发
     * publisher returnCallback  e端为成功投递到queue退回模式
     * consumer ack  消费端手动ack，处理一个消息,移除一个防止由于消费者宕机,
     * 还有一种解决方案,可以使用事务消息,性能下降250倍,不使用,了解即可。为此引入确认机制。
     */




























}
